package plugin.validator;

import java.math.BigDecimal;

import org.pentaho.di.core.Const;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowDataUtil;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;

public class PeselValidatorStep extends BaseStep implements StepInterface {

	private static Class<?> PKG = PeselValidatorStepMeta.class;

	PeselValidatorStepData data;
	PeselValidatorStepMeta meta;

	public PeselValidatorStep(StepMeta stepMeta,
			StepDataInterface stepDataInterface, int copyNr,
			TransMeta transMeta, Trans trans) {
		super(stepMeta, stepDataInterface, copyNr, transMeta, trans);

	}

	public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
		meta = (PeselValidatorStepMeta) smi;
		data = (PeselValidatorStepData) sdi;

		if (super.init(smi, sdi)) {
			if (Const.isEmpty(meta.getBirthDateFieldName())
					|| Const.isEmpty(meta.getLogicalFieldName())
					|| Const.isEmpty(meta.getSexFieldName())) {
				logError("ResultFieldMissing");
				return false;
			}
			return true;
		}
		return false;
	}

	public String toString() {
		return this.getClass().getName();
	}

	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi)
			throws KettleException {

		meta = (PeselValidatorStepMeta) smi;
		data = (PeselValidatorStepData) sdi;

		boolean sendToErrorRow = false;
		String errorMessage = null;

		Object[] r = getRow(); // Get row from input rowset & set row busy!
		if (r == null) { // no more input to be expected...
			setOutputDone();
			return false;
		}

		boolean isCorrect = false;
		String birthDate = null;
		String sex = null;

		if (first) {
			first = false;

			// get the RowMeta
			data.previousRowMeta = getInputRowMeta().clone();
			data.nrPrevFields = data.previousRowMeta.size();
			data.outputRowMeta = data.previousRowMeta;
			meta.getFields(data.outputRowMeta, getStepname(), null, null, this);

			// Check if field is provided
			if (Const.isEmpty(meta.getProcessedFieldName())) {
				logError(BaseMessages.getString(PKG,
						"PeselValidatorStepMeta.CheckResult.PeselFieldMissing"));
				throw new KettleException(BaseMessages.getString(PKG,
						"PeselValidatorStepMeta.CheckResult.PeselFieldMissing"));
			}

			// cache the position of the field
			if (data.indexOfField < 0) {
				data.indexOfField = getInputRowMeta().indexOfValue(
						meta.getProcessedFieldName());
				if (data.indexOfField < 0) {
					// The field is unreachable !
					throw new KettleException(
							BaseMessages
									.getString(
											PKG,
											"PeselValidatorStep.Exception.CouldnotFindField", meta.getProcessedFieldName())); //$NON-NLS-1$ //$NON-NLS-2$
				}
			}
			data.realLogicalFieldName = environmentSubstitute(meta
					.getLogicalFieldName());
			data.realBirthDateFieldName = environmentSubstitute(meta
					.getBirthDateFieldName());
			data.realSexFieldName = environmentSubstitute(meta
					.getSexFieldName());

		}// End If first

		Object[] outputRow = RowDataUtil.allocateRowData(data.outputRowMeta
				.size());
		for (int i = 0; i < data.nrPrevFields; i++) {
			outputRow[i] = r[i];
		}
		try {
			// get field
			
				
			Long pesel = getInputRowMeta().getInteger(r, data.indexOfField);
			String fieldvalue = pesel.toString();
			
			fieldvalue = pesel.toString();
			Result res = new Result();
			res = Validator.verifyPesel(fieldvalue);
			// Check if Card is Valid?
			isCorrect = res.isCorrect();
			// include Card Type?
			if (!Const.isEmpty(data.realBirthDateFieldName))
				birthDate = res.getBirthDate().toString();
			// include Not valid message?
			if (!Const.isEmpty(data.realSexFieldName))
				sex = res.getSex();
			
			// add card is Valid
			outputRow[data.nrPrevFields] = isCorrect;
			int rowIndex = data.nrPrevFields;
			rowIndex++;
			
			// add card type?
			if (!Const.isEmpty(data.realBirthDateFieldName)) {
				outputRow[rowIndex++] = birthDate;
			}
			// add not valid message?
			if (!Const.isEmpty(data.realSexFieldName)) {
				outputRow[rowIndex++] = sex;
			}
			
			// add new values to the row.
			putRow(data.outputRowMeta, outputRow); // copy row to output
													// rowset(s);

			if (log.isRowLevel())
				logRowlevel(BaseMessages.getString(PKG,
						"PeselValidatorStep.LineNumber", getLinesRead()
								+ " : " + getInputRowMeta().getString(r)));

		} catch (Exception e) {
			if (getStepMeta().isDoingErrorHandling()) {
				sendToErrorRow = true;
				errorMessage = e.toString();
			} else {
				logError(BaseMessages.getString(PKG,
						"PeselValidatorStep.ErrorInStepRunning") + e.getMessage()); //$NON-NLS-1$
				setErrors(1);
				stopAll();
				setOutputDone(); // signal end to receiver(s)
				return false;
			}
			if (sendToErrorRow) {
				// Simply add this row to the error row
				putError(getInputRowMeta(), r, 1, errorMessage,
						meta.getProcessedFieldName(), "PeselValidator001");
			}
		}

		return true;
	}
}
